// Package master : 主节点模块
// 实现主节点的基础功能
package master

import (
	"bufio"
	"context"
	"distributed_db/utils"
	"distributed_db/utils/consistent_hash"
	"distributed_db/utils/db"
	"distributed_db/utils/safe_queue"
	"distributed_db/utils/transaction"
	"fmt"
	"log"
	"os"
	"strings"
	"sync"
	"time"

	_ "github.com/go-sql-driver/mysql"
)

// RegionServerFile 存储 Region Server 的文件名
// TableFile        存储表列表的文件名
// VirtualNodes 	表示虚拟节点的数量，即每个 RegionServer 生成虚拟节点的数量
// MaxAttempts  	表示尝试选择 RegionServer 的最大次数
// RegionSize   	表示每个 Region 的大小，即存储的数据条数
const (
	RegionServerFile = "region_servers.txt"
	TableFile        = "tables"
	VirtualNodes     = 1000
	MaxAttempts      = ^uint32(0)
	RegionSize       = 100
	CheckInterval    = 60 * time.Second
	MaxRetry         = 5
	RetryInterval    = 1 * time.Second
)

// RegionServer 表示一个 Region Server，每个 Region Server 由 IP 和 Port 两个要素决定
// Regions RegionServer 中管理的 Regions
// Conn 和 RegionServer 的 RPC 连接对象
type RegionServer struct {
	ID      string
	IP      string
	Port    string
	Regions []db.Region
	Conn    *utils.ClientEnd
}

// RegionMapping 表示对一个 Region 的映射关系
// 每个 Region 都被对应的 RegionMapping 唯一表示
type RegionMapping struct {
	RegionName     string
	RecordNum      int
	RegionServerID string
}

// Table 表示一个表及其对应的 Region 映射关系
// TableName 表名
// Columns 该表中的列
// PrimaryKey 表中的主键
// Indexes 该表的索引
// RecordNum 该表中记录的数量
// Regions 存储该表的的 Region 信息
type Table struct {
	TableName  string
	Columns    []string
	PrimaryKey []string
	Indexes    []string
	RecordNum  int
	Regions    []RegionMapping
}

// Master 存储了 Master 运行所需的变量
type Master struct {
	name               string
	regionServers      []RegionServer
	tables             map[string]*Table
	transactions       map[string]*transaction.Transaction // 事务列表
	consistentHash     *consistent_hash.ConsistentHash     // 一致性哈希（哈希环）对象
	taskQueue          *safe_queue.SafeQueue               // 任务队列，用于缓存 regionServers 和 tables 的持久化任务
	taskWg             sync.WaitGroup                      // 任务队列的等待组
	regionServersMutex sync.Mutex                          // 互斥锁，保证 regionServers 的线程安全
	tablesMutex        sync.Mutex                          // 互斥锁，保证 tables 的线程安全
	ctx                context.Context                     // 管理协程的上下文
	cancel             context.CancelFunc                  // 取消上下文
}

// tablesToFile 将表列表同步到文件
func tablesToFile(master *Master) {
	master.tablesMutex.Lock()
	defer master.tablesMutex.Unlock()

	file, err := os.Create(TableFile + master.name + ".txt")
	if err != nil {
		log.Printf("Failed to create file: %v", err)
		return
	}
	defer func() {
		if closeErr := file.Close(); closeErr != nil {
			log.Printf("Failed to close file: %v", closeErr)
		}
	}()

	writer := bufio.NewWriter(file)
	for _, table := range master.tables {
		// 写入 TableName
		line := fmt.Sprintf("%s ", table.TableName)
		// 写入 Columns
		line += fmt.Sprintf("%d ", len(table.Columns))
		for _, col := range table.Columns {
			line += fmt.Sprintf("%s ", col)
		}
		// 写入 PrimaryKey
		line += fmt.Sprintf("%d ", len(table.PrimaryKey))
		for _, key := range table.PrimaryKey {
			line += fmt.Sprintf("%s ", key)
		}
		// 写入 Indexes
		line += fmt.Sprintf("%d ", len(table.Indexes))
		for _, index := range table.Indexes {
			line += fmt.Sprintf("%s ", index)
		}
		// 写入换行符
		line += "\n"
		_, err := writer.WriteString(line)
		if err != nil {
			log.Printf("Failed to write to file: %v", err)
			return
		}
	}
	// 确保所有缓冲区内容被刷新到文件中
	if err := writer.Flush(); err != nil {
		log.Printf("Failed to flush writer: %v", err)
	}
}

// regionServersToFile 同步 RegionServer 列表到文件
func regionServersToFile(master *Master) {
	master.regionServersMutex.Lock()
	defer master.regionServersMutex.Unlock()

	file, err := os.Create(RegionServerFile)
	if err != nil {
		log.Printf("Failed to create file: %v", err)
		return
	}
	defer func() {
		if closeErr := file.Close(); closeErr != nil {
			log.Printf("Failed to close file: %v", closeErr)
		}
	}()

	writer := bufio.NewWriter(file)
	for _, rs := range master.regionServers {
		// 修改写入逻辑，包含 ID 字段
		_, err := writer.WriteString(fmt.Sprintf("%s,%s,%s\n", rs.ID, rs.IP, rs.Port))
		if err != nil {
			log.Printf("Failed to write to file: %v", err)
			return
		}
	}

	// 确保所有缓冲区内容被刷新到文件中
	if err := writer.Flush(); err != nil {
		log.Printf("Failed to flush writer: %v", err)
	}
}

// loadTablesFromFile 从文件加载表列表
func (master *Master) loadTablesFromFile(filePath string) error {
	file, err := os.Open(filePath + master.name)
	if err != nil {
		if os.IsNotExist(err) {
			// 如果文件不存在，认为没有表需要加载
			return nil
		}
		return fmt.Errorf("failed to open file: %v", err)
	}
	defer func() {
		if closeErr := file.Close(); closeErr != nil {
			log.Printf("Failed to close file: %v", closeErr)
		}
	}()

	master.tablesMutex.Lock()
	defer master.tablesMutex.Unlock()

	scanner := bufio.NewScanner(file)
	for scanner.Scan() {
		line := strings.TrimSpace(scanner.Text())
		if line == "" {
			continue
		}
		parts := strings.Fields(line)
		idx := 0
		// 解析 TableName
		tableName := parts[idx]
		idx++
		// 解析 Columns
		columnCount := parseInt(parts[idx])
		idx++
		columns := parts[idx : idx+columnCount]
		idx += columnCount
		// 解析 PrimaryKey
		keyCount := parseInt(parts[idx])
		idx++
		primaryKey := parts[idx : idx+keyCount]
		idx += keyCount
		// 解析 Indexes
		indexCount := parseInt(parts[idx])
		idx++
		indexes := parts[idx : idx+indexCount]
		// 初始化表对象并加载到全局变量中
		master.tables[tableName] = &Table{
			TableName:  tableName,
			Columns:    columns,
			PrimaryKey: primaryKey,
			Indexes:    indexes,
			RecordNum:  0,
			Regions:    []RegionMapping{},
		}
	}

	if err := scanner.Err(); err != nil {
		return fmt.Errorf("failed to read file: %v", err)
	}

	return nil
}

// loadRegionServersFromFile 从本地文件加载 Region Server 列表
func (master *Master) loadRegionServersFromFile(filePath string) error {
	// 检查文件是否存在，如果不存在则创建一个空文件
	if _, err := os.Stat(filePath); os.IsNotExist(err) {
		file, err := os.Create(filePath)
		if err != nil {
			return err
		}
		defer func(file *os.File) {
			err := file.Close()
			if err != nil {
				log.Fatal(err)
			}
		}(file)
	}

	file, err := os.Open(filePath)
	if err != nil {
		return err
	}
	defer func(file *os.File) {
		err := file.Close()
		if err != nil {
			log.Fatal(err)
		}
	}(file)

	scanner := bufio.NewScanner(file)
	for scanner.Scan() {
		line := scanner.Text()
		parts := strings.Split(line, ",")
		if len(parts) != 3 {
			log.Printf("Invalid line in file: %s", line)
			continue
		}
		regionServer := RegionServer{
			ID:   parts[0],
			IP:   parts[1],
			Port: parts[2],
		}
		regionServer.Conn = utils.MakeEnd(fmt.Sprintf("%s:%s", regionServer.IP, regionServer.Port))
		// 如果 Conn 对象为 nil，则说明连接失败
		if regionServer.Conn != nil {
			master.regionServers = append(master.regionServers, regionServer)
			// 将新节点添加到一致性哈希环中
			serverKey := fmt.Sprintf("%s:%s", regionServer.IP, regionServer.Port)
			master.consistentHash.AddNode(serverKey)
		}
	}

	if err := scanner.Err(); err != nil {
		return err
	}

	return nil
}

// MaterInit 初始化 master 节点，返回一个可供操作的 Master 结构体
func MaterInit(name string) *Master {
	var err error
	// 初始化上下文和取消函数
	ctx, cancel := context.WithCancel(context.Background())
	// 初始化 master 节点
	master := &Master{
		name:           name,
		tables:         make(map[string]*Table),
		consistentHash: consistent_hash.NewConsistentHash(VirtualNodes),
		taskQueue:      safe_queue.NewSafeQueue(),
		ctx:            ctx,
		cancel:         cancel,
	}
	// 从本地文件加载 Region Server 列表
	err = master.loadRegionServersFromFile(RegionServerFile)
	// 加载 Table 列表
	err = master.loadTablesFromFile(TableFile)
	// 获取所有的 Region 信息，并构造 Region Server 列表和 Table 列表
	err = master.fetchRegions()
	if err != nil {
		log.Fatal(err)
	}
	// 将所有 RegionServer 添加到一致性哈希环中
	for _, server := range master.regionServers {
		serverKey := fmt.Sprintf("%s:%s", server.IP, server.Port)
		master.consistentHash.AddNode(serverKey)
	}
	// 通过协程开启心跳检测机制
	master.taskWg.Add(1)
	go master.keepalive()
	// 启动任务队列协程
	master.taskWg.Add(1)
	go master.runTaskCoroutine()
	return master
}

// Shutdown 释放 master 模块中的所有资源，并确保最后一次同步 RegionServer 列表到文件
func (master *Master) Shutdown() {
	fmt.Println("Master Shutdown Call")
	// 确保最后一次同步 RegionServer 列表到文件
	master.taskQueue.Enqueue(func() {
		regionServersToFile(master)
	})
	// 确保最后一次同步 Table 列表到文件
	master.taskQueue.Enqueue(func() {
		tablesToFile(master)
	})

	// 关闭任务队列
	master.taskQueue.Close()

	// 取消上下文，通知所有协程停止
	master.cancel()

	// 等待所有协程退出
	// 等待任务队列协程退出
	master.taskWg.Wait()
	// 最后一次同步，确保数据可靠
	fmt.Println("regionServers is: ", master.regionServers)
	fmt.Println("tables is: ", master.tables)
	regionServersToFile(master)
	tablesToFile(master)
}

// keepalive 维持所有 RegionServer 的心跳
func (master *Master) keepalive() {
	defer master.taskWg.Done()
	for {
		select {
		case <-master.ctx.Done():
			return
		default:
			select {
			case <-master.ctx.Done():
				return
			case <-time.After(CheckInterval):
				master.regionServersMutex.Lock()
				regionServers := make([]RegionServer, len(master.regionServers))
				copy(regionServers, master.regionServers)
				master.regionServersMutex.Unlock()

				for _, rs := range regionServers {
					go master.checkRegionServer(rs)
				}
			}
		}
	}
}

// runTaskCoroutine 处理 Region Server 的同步请求队列
func (master *Master) runTaskCoroutine() {
	defer master.taskWg.Done()
	for {
		select {
		case <-master.ctx.Done():
			fmt.Println("Out runTaskCoroutine exist")
			return
		default:
			select {
			case <-master.ctx.Done():
				fmt.Println("In runTaskCoroutine exist")
				return
			default:
				// 检查队列是否为空
				for master.taskQueue.Length() == 0 && !master.taskQueue.Closed {
					master.taskQueue.Cond.L.Lock()
					master.taskQueue.Cond.Wait()
					master.taskQueue.Cond.L.Unlock()
				}

				// 获取任务并执行
				task, ok := master.taskQueue.Front()
				if !ok {
					continue
				}
				if taskFunc, ok := task.(func()); ok {
					taskFunc()
				}
				master.taskQueue.Dequeue()
			}
		}
	}
}

// checkRegionServer 检查单个 RegionServer 的健康状态
func (master *Master) checkRegionServer(rs RegionServer) {
	var success bool
	for attempt := 1; attempt <= MaxRetry; attempt++ {
		// 尝试连接 RegionServer
		conn := utils.MakeEnd(fmt.Sprintf("%s:%s", rs.IP, rs.Port))
		if conn != nil {
			success = true
			break
		}
		// 连接失败，等待 RetryInterval 的时间后重试
		time.Sleep(RetryInterval)
	}

	if !success {
		// 如果 MaxRetry 次连接均失败，移除该 Region Server
		err := master.RemoveRegionServer(rs.IP, rs.Port)
		if err != nil {
			fmt.Println(err)
		}
	}
}

// GetRegionServers 获取 master 中的 RegionServer List
func (master *Master) GetRegionServers() []RegionServer {
	return master.regionServers
}

// GetTables 获取 master 中的 Table List
func (master *Master) GetTables() map[string]*Table {
	return master.tables
}

// parseInt 辅助函数：将字符串转换为整数
func parseInt(s string) int {
	var num int
	_, err := fmt.Sscanf(s, "%d", &num)
	if err != nil {
		return 0
	}
	return num
}
